package dm.ass2.eval;

import java.util.List;
import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import dm.ass1.io.ProfileWritable;
import dm.ass2.ClusteringDriver;
import dm.ass2.io.CodebookItemsWritable;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.user.UserMap;
import dm.ass2.eval.io.*;
public class EvalMapper extends Mapper<LongWritable, Text, IntWritable, ErrorWritable> {
	List<CodebookItemsWritable> cbItems;
	List<CodebookUsersWritable> cbUsers;
	private static final Log LOG = LogFactory.getLog(EvalMapper.class);
	@Override
	protected void map(LongWritable key, Text value1, Context context) throws IOException, InterruptedException {
		ProfileWritable value = ProfileWritable.fromString(value1.toString());
		int total = value.getRated().length;
		double error = 0;
		int userId = value.getRater();
		int[] items = value.getRated();
		int[] ratings = value.getRating();
		CodebookUsersWritable userClusterItem = findRatingUser(userId);
		int userCluster = userClusterItem.getUserClusterId();//getting the user cluster
		int totalmiss = 0;
		for (int itemIndex = 0; itemIndex < items.length; itemIndex++){
			CodebookItemsWritable itemCluster = findRatingItem(items[itemIndex]);
      	    if (itemCluster!=null){ 	
    	      double clusterRating = itemCluster.getUserClusterRating()[userCluster];
    	      if (!Double.isNaN(clusterRating)){
    	      error = error + Math.pow(ratings[itemIndex]-clusterRating, 2);
    	      }
    	      else{
    	    	  total = total - 1;
    	    	  System.out.println("saf");
    	      }
    	      }
    	    else {
    		  total = total - 1;
    		  totalmiss=totalmiss +1;
		         }
      	    }
		int totals[] = new int[1];
		double errors[] = new double[1];
		totals[0]= total;
		errors[0] = error;
      	context.write(new IntWritable(1),new ErrorWritable(1,totals,errors));
	}

	private CodebookItemsWritable findRatingItem(int item) {
		for (CodebookItemsWritable cbItem : cbItems) {
			if (ArrayUtils.contains(cbItem.getItemIds(), item)) {
				return cbItem;
			}
		}
		return null;
	}
	private CodebookUsersWritable findRatingUser(int item) {
		for (CodebookUsersWritable cbUser : cbUsers) {
			if (ArrayUtils.contains(cbUser.getUsrIds(), item)) {
				return cbUser;
			}
		}
		return null;
	}
	

	/**
	 * Initialization: loads the CodeBookItem,CodeBookUser to the memory.
	 */
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		readItems(context);
		readUsers(context);
		/*
		Path cbItemsPath = new Path("cbItems");
		Path cbUserPath = new Path("cbUsers");
		FileSystem fs = FileSystem.get(context.getConfiguration());
		cbItems = new ArrayList<CodebookItemsWritable>();
        cbUsers = new ArrayList<CodebookUsersWritable>();
		FSDataInputStream inItem = fs.open(cbItemsPath);
		FSDataInputStream inUser = fs.open(cbUserPath);
		try {
			CodebookItemsWritable cbItem = new CodebookItemsWritable();
			CodebookUsersWritable cbUser = new CodebookUsersWritable();
			cbItem.readFields(inItem);
			while (cbItem != null) {
				cbItems.add(cbItem);
				cbItem = new CodebookItemsWritable();
				cbItem.readFields(inItem);				
			}
			
			cbUser.readFields(inUser);			
			while (cbUser != null){
				cbUsers.add(cbUser);
				cbUser = new CodebookUsersWritable();
				cbUser.readFields(inUser);
			}
		}
			catch (EOFException e) {
				LOG.info(String.format("Setup: read %d records to cb item.", cbUsers.size()));
		    }
			finally {
			inItem.close();
			inUser.close();
		}
		*/
	}
	protected void readItems(Context context) throws IOException, InterruptedException {
		//super.setup(context);
		Path cbItemsPath = new Path(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_ITEMS_FILENAME);
		FileSystem fs = FileSystem.get(context.getConfiguration());
		cbItems = new ArrayList<CodebookItemsWritable>();
		FSDataInputStream inItem = fs.open(cbItemsPath);
		try {
			CodebookItemsWritable cbItem = new CodebookItemsWritable();
			cbItem.readFields(inItem);
			while (cbItem != null) {
				cbItems.add(cbItem);
				cbItem = new CodebookItemsWritable();
				cbItem.readFields(inItem);				
			}
		}
			catch (EOFException e) {
				LOG.info(String.format("Setup: read %d records to cb item.", cbItems.size()));
		    }
			finally {
			inItem.close();
		}
	}
	protected void readUsers(Context context) throws IOException, InterruptedException {
		//super.setup(context);
		System.out.println(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_USERS_FILENAME);
		Path cbUserPath = new Path(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_USERS_FILENAME);
		FileSystem fs = FileSystem.get(context.getConfiguration());
        cbUsers = new ArrayList<CodebookUsersWritable>();
		FSDataInputStream inUser = fs.open(cbUserPath);
		try {
			CodebookUsersWritable cbUser = new CodebookUsersWritable();		
			cbUser.readFields(inUser);			
			while (cbUser != null){
				cbUsers.add(cbUser);
				cbUser = new CodebookUsersWritable();
				cbUser.readFields(inUser);
			}
		}
			catch (EOFException e) {
				LOG.info(String.format("Setup: read %d records to cb item.", cbUsers.size()));
		    }
			finally {
			inUser.close();
		}
	}
	
	
	
	
	
}
